package com.atguigu.tabletest

import com.atguigu.api.SensorReading
import org.apache.flink.streaming.api._
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.Table
import org.apache.flink.table.api.scala._

/**
 *
 * @description: 在processTime处理模式下,对延时数据可以直接将时间延时若干单位时间,
 *               在eventTime处理模式下,对延时数据就可以通过watermark机制来处理,watermark用于表示timestamp小于Watermark的数据都已经达到.用来让程序自己平衡延迟和结果正确性
 *               watermark是一条特殊的数据记录
 *               必须单调递增,以确保任务的事件时间时钟都在向前推进,而不是在后退,watermark与数据的时间戳相关,
 *               首先确定延时3s[数据中最大差值],0-5s的窗口 6-10 首先数据都各自放到相应
 *               的桶里面,然后来了8 8-3=5那么0-5这个窗口就可以关闭计算了,后来的4则丢弃
 * @time: 2021-03-15 15:35
 * @author: baojinlong 
 **/
object Example01 {
  def main(args: Array[String]): Unit = {
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 设置并行度
    environment.setParallelism(1)
    //设置事件时间机制,watermark仅仅在事件时间语义下处理策略
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    // 设置生成watermark周期时间
    environment.getConfig.setAutoWatermarkInterval(500)
    // 从文本读取
    val inputStreamFromFile: DataStream[String] = environment.readTextFile("E:/big-data/FlinkTutorial/src/main/resources/sensor.data")
    // 基本转换操作
    val dataStream: DataStream[SensorReading] = inputStreamFromFile
      .map(data => {
        val dataArray: Array[String] = data.split(",")
        SensorReading(dataArray(0), dataArray(1).toLong, dataArray(2).toDouble)
      })

    // 首先创建表执行环境
    val tableEnvironment: StreamTableEnvironment = StreamTableEnvironment.create(environment)
    // 基于流创建一张表
    val dataTable: Table = tableEnvironment.fromDataStream(dataStream)
    val resultTable: Table = dataTable
      .select("id,temperature")
      .filter("id== 'sensor_1'")
    resultTable.toAppendStream[(String, Double)].print("resultTable")

    // 直接用sql实现resultSqlTable
    tableEnvironment.createTemporaryView("dataTable", dataTable)
    val sql: String = "select id,temperatue from dataTable where id='sensor_1'"
    val resultSqlTable: Table = tableEnvironment.sqlQuery(sql)
    resultSqlTable.toAppendStream[(String, Double)].print("resultSqlTable")

    environment.execute("xx")
  }

}
